package bloom

import (
	"crypto/md5"
	"database/sql"
	"encoding/binary"
	"fmt"
	"gitee.com/tziye/renai-go/common"
	"gitee.com/tziye/renai-go/renai"
	"gitee.com/tziye/renai-go/sqlproxy"
	"math"
	"strings"
)

type BloomOperator struct {
	RenaiClient *renai.RenaiClient
	Db          *sql.DB
}

func NewBloomOperator(renaiClient *renai.RenaiClient) *BloomOperator {
	return &BloomOperator{
		RenaiClient: renaiClient,
		Db:          renaiClient.Db,
	}
}

func (bloomOperator *BloomOperator) Put(module string, value any) bool {
	hashList := calculate(value)
	s := "INSERT INTO renai_bloom_filter(module, %s) VALUES (?, %s) ON DUPLICATE KEY UPDATE %s, update_time = NOW()"
	l1 := make([]string, 0)
	l2 := make([]string, 0)
	l3 := make([]string, 0)
	for _, t := range hashList {
		l1 = append(l1, "h"+common.String(t.V1))
		l2 = append(l2, common.String(t.V2))
		l3 = append(l3, fmt.Sprintf("h%d = h%d | %d", t.V1, t.V1, t.V2))
	}
	s1 := strings.Join(l1, ",")
	s2 := strings.Join(l2, ",")
	s3 := strings.Join(l3, ",")
	s = fmt.Sprintf(s, s1, s2, s3)
	rows := sqlproxy.Exec(bloomOperator.Db, s, module)
	return rows > 0
}

func (bloomOperator *BloomOperator) MightContain(module string, value any) bool {
	hashList := calculate(value)
	l1 := make([]string, 0)
	for _, t := range hashList {
		l1 = append(l1, fmt.Sprintf(" h%d & %d > 0 ", t.V1, t.V2))
	}
	s1 := strings.Join(l1, "AND")
	s := "SELECT id FROM renai_bloom_filter WHERE module = ? AND %s"
	s = fmt.Sprintf(s, s1)
	row := sqlproxy.QueryRow(bloomOperator.Db, s, module)
	var id sql.NullInt64
	err := row.Scan(&id)
	if common.LogErr(err) {
		return false
	}
	return id.Valid
}

func (bloomOperator *BloomOperator) Delete(module string) bool {
	s := "DELETE FROM renai_bloom_filter WHERE module = ?"
	rows := sqlproxy.Exec(bloomOperator.Db, s, module)
	return rows > 0
}

const (
	longAddressableBits = 6
	bitSize             = 258816
	numHashFunctions    = 10
)

func calculate(value any) []common.Tuple2 {
	bytes := hash(value)
	hash1 := lowerEight(bytes)
	hash2 := upperEight(bytes)

	list := make([]common.Tuple2, 0)
	set := make(map[uint64]bool)

	combinedHash := hash1
	for i := 0; i < numHashFunctions; i++ {
		bitIndex := (combinedHash & math.MaxUint64) % bitSize
		longIndex := bitIndex >> longAddressableBits
		var mask uint64 = 1 << (bitIndex % 64)
		if !set[longIndex] {
			list = append(list, common.NewTuple2(longIndex, mask))
			set[longIndex] = true
		}
		combinedHash += hash2
	}
	return list
}

func hash(value any) [16]byte {
	str := common.String(value)
	return md5.Sum([]byte(str))
}

func lowerEight(bytes [16]byte) uint64 {
	byteData := []byte{bytes[7], bytes[6], bytes[5], bytes[4], bytes[3], bytes[2], bytes[1], bytes[0]}
	return binary.BigEndian.Uint64(byteData)
}

func upperEight(bytes [16]byte) uint64 {
	byteData := []byte{bytes[15], bytes[14], bytes[13], bytes[12], bytes[11], bytes[10], bytes[9], bytes[8]}
	return binary.BigEndian.Uint64(byteData)
}
